home *** CD-ROM | disk | FTP | other *** search
/ PC World Komputer 2010 April / PCWorld0410.iso / hity wydania / Ubuntu 9.10 PL / karmelkowy-koliberek-9.10-netbook-remix-PL.iso / casper / filesystem.squashfs / usr / share / pyshared / nevow / testutil.py < prev    next >
Encoding:
Python Source  |  2008-02-06  |  14.9 KB  |  538 lines
  1. # Copyright (c) 2004 Divmod.
  2. # See LICENSE for details.
  3.  
  4. import os, sys, signal
  5.  
  6. try:
  7.     from popen2 import Popen3
  8. except ImportError:
  9.     Popen3 = None
  10.  
  11. from zope.interface import implements
  12.  
  13. try:
  14.     import subunit
  15. except ImportError:
  16.     subunit = None
  17.  
  18. from twisted.trial.unittest import TestCase as TrialTestCase
  19. from twisted.python.components import Componentized
  20. from twisted.internet import defer
  21. from twisted.web import http
  22. from twisted.protocols.basic import LineReceiver
  23.  
  24. from formless import iformless
  25.  
  26. from nevow import inevow, context, athena, loaders, tags, appserver
  27. from nevow.jsutil import findJavascriptInterpreter, generateTestScript
  28.  
  29. class FakeChannel:
  30.     def __init__(self, site):
  31.         self.site = site
  32.  
  33.  
  34. class FakeSite:
  35.     pass
  36.  
  37.  
  38. class FakeSession(Componentized):
  39.     implements(inevow.ISession)
  40.     def __init__(self, avatar):
  41.         Componentized.__init__(self)
  42.         self.avatar = avatar
  43.         self.uid = 12345
  44.     def getLoggedInRoot(self):
  45.         return self.avatar
  46.  
  47.  
  48. fs = FakeSession(None)
  49.  
  50.  
  51. class FakeRequest(Componentized):
  52.     """
  53.     Implementation of L{inevow.IRequest} which is convenient to use in unit
  54.     tests.
  55.  
  56.     @ivar lastModified: The value passed to L{setLastModified} or C{None} if
  57.         that method has not been called.
  58.  
  59.     @type accumulator: C{str}
  60.     @ivar accumulator: The bytes written to the response body.
  61.  
  62.     @type deferred: L{Deferred}
  63.     @ivar deferred: The deferred which represents rendering of the response
  64.         to this request.  This is basically an implementation detail of
  65.         L{NevowRequest}.  Application code should probably never use this.
  66.  
  67.     @ivar _appRootURL: C{None} or the object passed to L{rememberRootURL}.
  68.     """
  69.     implements(inevow.IRequest)
  70.  
  71.     fields = None
  72.     failure = None
  73.     context = None
  74.     redirected_to = None
  75.     lastModified = None
  76.     content = ""
  77.     method = 'GET'
  78.     code = http.OK
  79.     deferred = None
  80.     accumulator = ''
  81.     _appRootURL = None
  82.  
  83.     def __init__(self, headers=None, args=None, avatar=None,
  84.                  uri='/', currentSegments=None, cookies=None,
  85.                  user="", password="", isSecure=False):
  86.         """Create a FakeRequest instance.
  87.  
  88.         headers:
  89.             dict of headers
  90.         args:
  91.             dict of args
  92.         avatar:
  93.             avatar to pass to the FakeSession instance
  94.         uri:
  95.             request URI
  96.         currentSegments:
  97.             list of segments that have "already been located"
  98.         cookies:
  99.             dict of cookies
  100.         user:
  101.             username (like in http auth)
  102.         password:
  103.             password (like in http auth)
  104.         isSecure:
  105.             whether this request represents an HTTPS url
  106.         """
  107.         Componentized.__init__(self)
  108.         self.uri = uri
  109.         self.prepath = []
  110.         postpath = uri.split('?')[0]
  111.         assert postpath.startswith('/')
  112.         self.postpath = postpath[1:].split('/')
  113.         if currentSegments is not None:
  114.             for seg in currentSegments:
  115.                 assert seg == self.postpath[0]
  116.                 self.prepath.append(self.postpath.pop(0))
  117.         else:
  118.             self.prepath.append('')
  119.         self.headers = {}
  120.         if headers:
  121.             for k, v in headers.iteritems():
  122.                 self.setHeader(k, v)
  123.         self.args = args or {}
  124.         self.sess = FakeSession(avatar)
  125.         self.site = FakeSite()
  126.         self.received_headers = {}
  127.         if cookies is not None:
  128.             self.cookies = cookies
  129.         else:
  130.             self.cookies = {}
  131.         self.user = user
  132.         self.password = password
  133.         self.secure = isSecure
  134.         self.deferred = defer.Deferred()
  135.  
  136.     def URLPath(self):
  137.         from nevow import url
  138.         return url.URL.fromString('')
  139.  
  140.     def getSession(self):
  141.         return self.sess
  142.  
  143.     def registerProducer(self, producer, streaming):
  144.         """
  145.         Synchronously cause the given producer to produce all of its data.
  146.  
  147.         This will not work with push producers.  Do not use it with them.
  148.         """
  149.         keepGoing = [None]
  150.         self.unregisterProducer = keepGoing.pop
  151.         while keepGoing:
  152.             producer.resumeProducing()
  153.         del self.unregisterProducer
  154.  
  155.     def v():
  156.         def get(self):
  157.             return self.accumulator
  158.         return get,
  159.     v = property(*v())
  160.  
  161.     def write(self, bytes):
  162.         """
  163.         Accumulate the given bytes as part of the response body.
  164.  
  165.         @type bytes: C{str}
  166.         """
  167.         self.accumulator += bytes
  168.  
  169.  
  170.     finished = False
  171.     def finishRequest(self, success):
  172.         self.finished = True
  173.  
  174.     def finish(self):
  175.         self.deferred.callback('')
  176.  
  177.     def getHeader(self, key):
  178.         return self.headers.get(key.lower())
  179.  
  180.     def setHeader(self, key, val):
  181.         self.headers[key.lower()] = val
  182.  
  183.     def redirect(self, url):
  184.         self.redirected_to = url
  185.  
  186.     def processingFailed(self, f):
  187.         self.failure = f
  188.  
  189.     def setResponseCode(self, code):
  190.         self.code = code
  191.  
  192.     def setLastModified(self, when):
  193.         self.lastModified = when
  194.  
  195.     def prePathURL(self):
  196.         """
  197.         The absolute URL up until the last handled segment of this request.
  198.  
  199.         @rtype: C{str}.
  200.         """
  201.         return 'http://%s/%s' % (self.getHeader('host') or 'localhost',
  202.                                  '/'.join(self.prepath))
  203.  
  204.     def getClientIP(self):
  205.         return '127.0.0.1'
  206.  
  207.     def addCookie(self, k, v, expires=None, domain=None, path=None, max_age=None, comment=None, secure=None):
  208.         """
  209.         Set a cookie for use in subsequent requests.
  210.         """
  211.         self.cookies[k] = v
  212.  
  213.     def getCookie(self, k):
  214.         """
  215.         Fetch a cookie previously set.
  216.         """
  217.         return self.cookies.get(k)
  218.  
  219.     def getUser(self):
  220.         """
  221.         Returns the HTTP auth username.
  222.         """
  223.         return self.user
  224.  
  225.     def getPassword(self):
  226.         """
  227.         Returns the HTTP auth password.
  228.         """
  229.         return self.password
  230.  
  231.     def getRootURL(self):
  232.         """
  233.         Return the previously remembered URL.
  234.         """
  235.         return self._appRootURL
  236.  
  237.  
  238.     def rememberRootURL(self, url=None):
  239.         """
  240.         For compatibility with appserver.NevowRequest.
  241.         """
  242.         if url is None:
  243.             raise NotImplementedError(
  244.                 "Default URL remembering logic is not implemented.")
  245.         self._appRootURL = url
  246.  
  247.  
  248.     def isSecure(self):
  249.         """
  250.         Returns whether this is an HTTPS request or not.
  251.         """
  252.         return self.secure
  253.  
  254.  
  255. class TestCase(TrialTestCase):
  256.     hasBools = (sys.version_info >= (2,3))
  257.     _assertions = 0
  258.  
  259.     # This should be migrated to Twisted.
  260.     def failUnlessSubstring(self, containee, container, msg=None):
  261.         self._assertions += 1
  262.         if container.find(containee) == -1:
  263.             self.fail(msg or "%r not in %r" % (containee, container))
  264.     def failIfSubstring(self, containee, container, msg=None):
  265.         self._assertions += 1
  266.         if container.find(containee) != -1:
  267.             self.fail(msg or "%r in %r" % (containee, container))
  268.  
  269.     assertSubstring = failUnlessSubstring
  270.     assertNotSubstring = failIfSubstring
  271.  
  272.     def assertNotIdentical(self, first, second, msg=None):
  273.         self._assertions += 1
  274.         if first is second:
  275.             self.fail(msg or '%r is %r' % (first, second))
  276.  
  277.     def failIfIn(self, containee, container, msg=None):
  278.         self._assertions += 1
  279.         if containee in container:
  280.             self.fail(msg or "%r in %r" % (containee, container))
  281.  
  282.     def assertApproximates(self, first, second, tolerance, msg=None):
  283.         self._assertions += 1
  284.         if abs(first - second) > tolerance:
  285.             self.fail(msg or "%s ~== %s" % (first, second))
  286.  
  287.  
  288. if not hasattr(TrialTestCase, 'mktemp'):
  289.     def mktemp(self):
  290.         import tempfile
  291.         return tempfile.mktemp()
  292.     TestCase.mktemp = mktemp
  293.  
  294.  
  295. class AccumulatingFakeRequest(FakeRequest):
  296.     """
  297.     I am a fake IRequest that is also a stub implementation of
  298.     IFormDefaults.
  299.  
  300.     This class is named I{accumulating} for historical reasons only.  You
  301.     probably want to ignore this and use L{FakeRequest} instead.
  302.     """
  303.     implements(iformless.IFormDefaults)
  304.  
  305.     def __init__(self, *a, **kw):
  306.         FakeRequest.__init__(self, *a, **kw)
  307.         self.d = defer.Deferred()
  308.  
  309.     def getDefault(self, key, context):
  310.         return ''
  311.  
  312.     def remember(self, object, interface):
  313.         pass
  314.  
  315.  
  316. class FragmentWrapper(athena.LivePage):
  317.     """
  318.     I wrap myself around an Athena fragment, providing a minimal amount of html
  319.     scaffolding in addition to an L{athena.LivePage}.
  320.     """
  321.     docFactory = loaders.stan(
  322.                     tags.html[
  323.                         tags.body[
  324.                             tags.directive('fragment')]])
  325.  
  326.     def __init__(self, f):
  327.         super(FragmentWrapper, self).__init__()
  328.         self.f = f
  329.  
  330.     def render_fragment(self, ctx, data):
  331.         self.f.setFragmentParent(self)
  332.         return self.f
  333.  
  334.  
  335. def renderLivePage(res, topLevelContext=context.WebContext,
  336.                    reqFactory=FakeRequest):
  337.     """
  338.     Render the given LivePage resource, performing LivePage-specific cleanup.
  339.     Return a Deferred which fires when it has rendered.
  340.     """
  341.     D = renderPage(res, topLevelContext, reqFactory)
  342.     return D.addCallback(lambda x: (res._messageDeliverer.close(), x)[1])
  343.  
  344.  
  345. def renderPage(res, topLevelContext=context.WebContext,
  346.                reqFactory=FakeRequest):
  347.     """
  348.     Render the given resource.  Return a Deferred which fires when it has
  349.     rendered.
  350.     """
  351.     req = reqFactory()
  352.     ctx = topLevelContext(tag=res)
  353.     ctx.remember(req, inevow.IRequest)
  354.  
  355.     render = appserver.NevowRequest(None, True).gotPageContext
  356.  
  357.     result = render(ctx)
  358.     result.addCallback(lambda x: req.accumulator)
  359.     return result
  360.  
  361.  
  362.  
  363. class NotSupported(Exception):
  364.     """
  365.     Raised by L{JavaScriptTestCase} if the installation lacks a certain
  366.     required feature.
  367.     """
  368.  
  369.  
  370.  
  371. class TestProtocolLineReceiverServer(LineReceiver):
  372.     """
  373.     Subunit protocol which is also a Twisted LineReceiver so that it
  374.     includes line buffering logic.
  375.     """
  376.     delimiter = '\n'
  377.  
  378.     def __init__(self, proto):
  379.         self.proto = proto
  380.  
  381.  
  382.     def lineReceived(self, line):
  383.         """
  384.         Forward the line on to the subunit protocol's lineReceived method,
  385.         which expects it to be newline terminated.
  386.         """
  387.         self.proto.lineReceived(line + '\n')
  388.  
  389.  
  390.  
  391. class JavaScriptTestCase(TrialTestCase):
  392.     def __init__(self, methodName='runTest'):
  393.         TrialTestCase.__init__(self, methodName)
  394.         self.testMethod = getattr(self, methodName)
  395.  
  396.  
  397.     def findJavascriptInterpreter(self):
  398.         """
  399.         @see L{nevow.jsutil.findJavascriptInterpreter}
  400.         """
  401.         return findJavascriptInterpreter()
  402.  
  403.  
  404.     def checkDependencies(self):
  405.         """
  406.         Check that all the dependencies of the test are satisfied.
  407.  
  408.         @raise NotSupported: If any one of the dependencies is not satisfied.
  409.         """
  410.         js = self.findJavascriptInterpreter()
  411.         if js is None:
  412.             raise NotSupported("Could not find JavaScript interpreter")
  413.         if subunit is None:
  414.             raise NotSupported("Could not import 'subunit'")
  415.         if Popen3 is None:
  416.             raise NotSupported("Could not import 'popen2.Popen3'")
  417.         for name in ['WEXITSTATUS', 'WIFSIGNALED' ,'WTERMSIG']:
  418.             if getattr(os, name, None) is None:
  419.                 raise NotSupported("os.%s unavailable" % (name,))
  420.  
  421.  
  422.     def _writeToTemp(self, contents):
  423.         fname = self.mktemp()
  424.         fd = file(fname, 'w')
  425.         try:
  426.             fd.write(contents)
  427.         finally:
  428.             fd.close()
  429.         return fname
  430.  
  431.  
  432.     def createSource(self, testModule):
  433.         """
  434.         Return a string of JavaScript source code which, when executed, will
  435.         run the JavaScript unit tests defined in the given module.
  436.  
  437.         @type testModule: C{str}
  438.         @param testModule: The JavaScript module name which contains the
  439.         tests to run.
  440.  
  441.         @rtype: C{str}
  442.         """
  443.         js = """
  444. // import Divmod.UnitTest
  445. // import %(module)s
  446.  
  447. Divmod.UnitTest.runRemote(Divmod.UnitTest.loadFromModule(%(module)s));
  448. """ % {'module': testModule}
  449.         return js
  450.  
  451.  
  452.     def makeScript(self, testModule):
  453.         """
  454.         Write JavaScript source for executing the JavaScript unit tests in
  455.         the given JavaScript module to a file and return the name of that
  456.         file.
  457.  
  458.         @type testModule: C{str}
  459.         @param testModule: The JavaScript module name which contains the
  460.         tests to run.
  461.  
  462.         @rtype: C{str}
  463.         """
  464.         jsfile = self._writeToTemp(self.createSource(testModule))
  465.         scriptFile = self._writeToTemp(generateTestScript(jsfile))
  466.         return scriptFile
  467.  
  468.  
  469.     def _runWithSigchild(self, f, *a, **kw):
  470.         """
  471.         Run the given function with an alternative SIGCHLD handler.
  472.         """
  473.         oldHandler = signal.signal(signal.SIGCHLD, signal.SIG_DFL)
  474.         try:
  475.             return f(*a, **kw)
  476.         finally:
  477.             signal.signal(signal.SIGCHLD, oldHandler)
  478.  
  479.  
  480.     def run(self, result):
  481.         try:
  482.             self.checkDependencies()
  483.         except NotSupported, e:
  484.             result.startTest(self)
  485.             result.addSkip(self, str(e))
  486.             result.stopTest(self)
  487.             return
  488.         js = self.findJavascriptInterpreter()
  489.         script = self.makeScript(self.testMethod())
  490.         server = subunit.TestProtocolServer(result)
  491.         protocol = TestProtocolLineReceiverServer(server)
  492.  
  493.         # What this *SHOULD BE*
  494.         # spawnProcess(protocol, js, (script,))
  495.         # return protocol.someDisconnectCallback()
  496.  
  497.         # However, *run cannot return a Deferred profanity profanity profanity
  498.         # profanity*, so instead it is *profanity* this:
  499.         def run():
  500.             child = Popen3([js, script])
  501.             while True:
  502.                 bytes = child.fromchild.read(4096)
  503.                 if bytes:
  504.                     protocol.dataReceived(bytes)
  505.                 else:
  506.                     break
  507.             exitCode = child.wait()
  508.             if os.WIFSIGNALED(exitCode):
  509.                 result.addError(
  510.                     self,
  511.                     (Exception,
  512.                      "JavaScript interpreter exited due to signal %d" % (os.WTERMSIG(exitCode),),
  513.                      None))
  514.             else:
  515.                 exitStatus = os.WEXITSTATUS(exitCode)
  516.                 if exitStatus:
  517.                     result.addError(
  518.                         self,
  519.                         (Exception,
  520.                          "JavaScript interpreter had error exit: %d" % (exitStatus,),
  521.                          None))
  522.         self._runWithSigchild(run)
  523.  
  524.  
  525.  
  526. def setJavascriptInterpreterOrSkip(testCase):
  527.     """
  528.     If we're unable to find a javascript interpreter (currently we only look
  529.     for smjs or js) then set the C{skip} attribute on C{testCase}. Otherwise
  530.     assign the path to the interpreter executable to
  531.     C{testCase.javascriptInterpreter}
  532.     """
  533.     script = findJavascriptInterpreter()
  534.     if script is None:
  535.         testCase.skip = "No JavaScript interpreter available."
  536.     else:
  537.         testCase.javascriptInterpreter = script
  538.